package com.acdc.mqtt.service;

import org.eclipse.paho.client.mqttv3.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import com.acdc.mqtt.config.MqttConfig;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.acdc.mqtt.service.impl.HeartUpHandler;
import com.acdc.mqtt.service.impl.WkPubHandler;
import com.acdc.mqtt.service.impl.OtaReqHandler;
import com.acdc.mqtt.service.impl._485AckHandler;

import com.acdc.mqtt.entity.MqttMsgType;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * MQTT消费者服务
 */
@Service
@Slf4j
public class MqttConsumerService {
    
    @Autowired
    @Qualifier("consumerMqttClient")
    private MqttClient mqttClient;
    
    @Autowired
    @Qualifier("consumerMqttConnectOptions")
    private MqttConnectOptions connectOptions;
    
    @Autowired
    private MqttConfig mqttConfig;

    @Autowired
    private HeartUpHandler heartUpHandler;

    @Autowired
    private _485AckHandler _485AckHandler;

    @Autowired
    private WkPubHandler wkPubHandler;

    @Autowired
    private OtaReqHandler otaReqHandler;
    
    private ExecutorService executorService;
    
    // 消息处理器映射
    private final ConcurrentHashMap<String, MessageHandler> messageHandlers = new ConcurrentHashMap<>();
    
    @PostConstruct
    public void init() throws InterruptedException {
        try {
            // 根据 MqttMsgType 将对应的handler 添加到 messageHandlers
            messageHandlers.put(MqttMsgType.HEART_UP.getMsgType(), heartUpHandler);
            messageHandlers.put(MqttMsgType._485_ACK.getMsgType(), _485AckHandler);
            messageHandlers.put(MqttMsgType._485_ACK.name(), _485AckHandler);
            messageHandlers.put(MqttMsgType.WK_PUB.getMsgType(), wkPubHandler);
            messageHandlers.put(MqttMsgType.OTA_REQ.getMsgType(), otaReqHandler);
            
            // 创建线程池
            executorService = Executors.newFixedThreadPool(10);
            
            // 配置离线消息接收
            configureOfflineMessageHandling();
            
            // 设置回调
            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    log.error("MQTT连接丢失，尝试重连...", cause);
                    // 自动重连
                    reconnectWithRetry();
                }
                
                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    handleMessage(topic, message);
                }
                
                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    log.debug("消息投递完成");
                }
            });
            
            // 连接到MQTT代理
            mqttClient.connect(connectOptions);
            log.info("MQTT消费者服务初始化成功，已配置离线消息接收");
            
            while(!mqttClient.isConnected()){
                Thread.sleep(1000);
                log.info("MQTT客户端未连接，等待重连...");
            }
            log.info("MQTT客户端已连接");
            // 订阅默认主题
            String defaultTopic = mqttConfig.rootTopic() + "+/UPLOAD";
            log.info("订阅主题: {}", defaultTopic);
            this.subscribeTopic(defaultTopic);
        } catch (MqttException e) {
            log.error("MQTT消费者服务初始化失败", e);
        }
    }
    
    /**
     * 配置离线消息接收
     */
    private void configureOfflineMessageHandling() {
        // 确保cleanSession为false，这样MQTT代理会保存会话状态和离线消息
        if (connectOptions.isCleanSession()) {
            log.warn("检测到cleanSession=true，这可能导致离线消息丢失！");
            log.info("建议在MqttConfig中设置cleanSession=false");
        }
        
        // 设置自动重连
        connectOptions.setAutomaticReconnect(true);
        connectOptions.setMaxReconnectDelay(10000); // 最大重连延迟10秒
        
        // 设置连接超时
        connectOptions.setConnectionTimeout(30);
        connectOptions.setKeepAliveInterval(60);
        
        log.info("离线消息配置: cleanSession={}, automaticReconnect={}, keepAlive={}s", 
                connectOptions.isCleanSession(), 
                connectOptions.isAutomaticReconnect(),
                connectOptions.getKeepAliveInterval());
    }
    
    /**
     * 重连重试机制
     */
    private void reconnectWithRetry() {
        int maxRetries = 5;
        int retryCount = 0;
        
        while (retryCount < maxRetries) {
            try {
                Thread.sleep(5000); // 等待5秒后重试
                if (!mqttClient.isConnected()) {
                    mqttClient.connect(connectOptions);
                    log.info("MQTT重连成功");
                    
                    // 重新订阅之前的主题
                    resubscribeTopics();
                    break;
                }
            } catch (Exception e) {
                retryCount++;
                log.error("MQTT重连失败，第{}次尝试: {}", retryCount, e.getMessage());
            }
        }
        
        if (retryCount >= maxRetries) {
            log.error("MQTT重连失败，已达到最大重试次数: {}", maxRetries);
        }
    }
    
    /**
     * 重新订阅之前的主题
     */
    private void resubscribeTopics() {
        try {
            // 重新订阅默认主题
            String defaultTopic = mqttConfig.rootTopic() + "+/UPLOAD";
            mqttClient.subscribe(defaultTopic, 1);
            log.info("重新订阅主题: {}", defaultTopic);
        } catch (MqttException e) {
            log.error("重新订阅主题失败", e);
        }
    }
    
    /**
     * 处理接收到的消息
     * @param topic 主题
     * @param message 消息
     */
    public void handleMessage(String topic, MqttMessage message) {
        String messageContent = new String(message.getPayload());
        log.info("收到消息: topic={}, message={}, qos={}", topic, messageContent, message.getQos());
        
        // 异步处理消息
        executorService.submit(() -> {
            try {
                // 将 messageContent 转为 JSON 对象
                JsonNode jsonNode = new ObjectMapper().readTree(messageContent);
                // 根据jsonNode的stream_id 获取对应的handler
                String streamId = jsonNode.get("stream_id").asText();
                MessageHandler handler = messageHandlers.get(streamId);
                
                if (handler != null) {
                    handler.handleMessage(topic, messageContent);
                } else {
                    log.warn("未找到主题 {} 的消息处理器", topic);
                }
            } catch (Exception e) {
                log.error("处理消息失败: topic={}, message={}", topic, messageContent, e);
            }
        });
    }
    
    /**
     * 订阅主题
     * @param topic 主题
     * @param qos 服务质量
     */
    public void subscribeTopic(String topic, int qos) {
        try {
            if (mqttClient != null && mqttClient.isConnected()) {
                mqttClient.subscribe(topic, qos);
                log.info("订阅主题成功: topic={}, qos={}", topic, qos);
            } else {
                log.warn("MQTT客户端未连接，无法订阅主题");
            }
        } catch (MqttException e) {
            log.error("订阅主题失败: topic={}", topic, e);
        }
    }
    
    /**
     * 订阅主题（默认QoS=1）
     * @param topic 主题
     * @param handler 消息处理器
     */
    public void subscribeTopic(String topic) {
        subscribeTopic(topic, 1);
    }

    public void subscribeTopic(MessageHandler handler) {
        // 从 MqttConfig 中获取 rootTopic
        String rootTopic = mqttConfig.rootTopic() + "+/UPLOAD";
        subscribeTopic(rootTopic, 1);
    }
    
    /**
     * 取消订阅主题
     * @param topic 主题
     */
    public void unsubscribeTopic(String topic) {
        try {
            if (mqttClient != null && mqttClient.isConnected()) {
                mqttClient.unsubscribe(topic);
                log.info("取消订阅主题成功: topic={}", topic);
            } else {
                log.warn("MQTT客户端未连接，无法取消订阅主题");
            }
        } catch (MqttException e) {
            log.error("取消订阅主题失败: topic={}", topic, e);
        }
    }
    
    /**
     * 检查连接状态
     * @return 是否已连接
     */
    public boolean isConnected() {
        return mqttClient != null && mqttClient.isConnected();
    }
    
    /**
     * 获取已订阅的主题数量
     * @return 主题数量
     */
    public int getSubscribedTopicCount() {
        return messageHandlers.size();
    }
    
    @PreDestroy
    public void destroy() {
        try {
            if (executorService != null) {
                executorService.shutdown();
            }
            
            if (mqttClient != null && mqttClient.isConnected()) {
                mqttClient.disconnect();
                mqttClient.close();
                log.info("MQTT消费者服务已关闭");
            }
        } catch (MqttException e) {
            log.error("关闭MQTT消费者服务失败", e);
        }
    }
} 